package com.atguigu.flink.datastreamapi.sink;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

/**
 * Created by Smexy on 2022/12/13
 *
 *     5大数据类型:
 *      string:   set k1 v1
 *      set:      sadd k1,v1,v2,v3
 *      zset:     zadd k1 score member...
 *      list:     lpush  k1 v1
 *                rpush  k1  v2
 *      hash:     hset  k1  field value
 *
 *    -----------------
 *      pubsub(发布订阅，类似kafka) ,hyperloglog(大基数统计)
 *
 *      --------
 *          输出的套路：  DataStream.addSink(SinkFunction s);
 */
public class Demo1_RedisSink
{

    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());


        // FlinkJedisConfigBase flinkJedisConfigBase
        FlinkJedisPoolConfig flinkJedisPoolConfig = new FlinkJedisPoolConfig.Builder()
            .setDatabase(0)
            .setHost("hadoop102")
            .setPort(6379)
            .setTimeout(60000)
            .setMaxIdle(10)     // 池子中最大空闲的连接数
            .setMinIdle(5)     // 池子中最小空闲的连接数
            .setMaxTotal(20)  // 池子容量
            .build();

        //RedisSink(FlinkJedisConfigBase flinkJedisConfigBase, RedisMapper<IN> redisSinkMapper)


        //输出到redis
        ds.addSink(new RedisSink<>(flinkJedisPoolConfig,new HashRedisMapper()));

        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }

    private static class HashRedisMapper implements RedisMapper<WaterSensor>{

        /*
                写命令

                 hash:     hset  k1(additionalKey)  field value
                RedisCommandDescription(RedisCommand redisCommand, String additionalKey)： 用于构造 hash 和 zset类型
                RedisCommandDescription(RedisCommand redisCommand)： 用于构造非 hash 和 zset类型
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET,"hashws");
        }

        /*
             field
         */
        @Override
        public String getKeyFromData(WaterSensor data) {
            return data.getId();
        }

        //value
        @Override
        public String getValueFromData(WaterSensor data) {
            return data.toString();
        }
    }


    private static class ZSetRedisMapper implements org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper<WaterSensor>
    {

        /*
                写命令

                  zset:     zadd k1(additionalKey) score member...
                RedisCommandDescription(RedisCommand redisCommand, String additionalKey)： 用于构造 hash 和 zset类型
                RedisCommandDescription(RedisCommand redisCommand)： 用于构造非 hash 和 zset类型
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.ZADD,"wszset");
        }

        /*
              member
         */
        @Override
        public String getKeyFromData(WaterSensor data) {
            return data.toString();
        }

        //score
        @Override
        public String getValueFromData(WaterSensor data) {
            return data.getVc().toString();
        }
    }

    private static class SetRedisMapper implements org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper<WaterSensor>
    {

        /*
                写命令

                RedisCommandDescription(RedisCommand redisCommand, String additionalKey)： 用于构造 hash 和 zset类型
                RedisCommandDescription(RedisCommand redisCommand)： 用于构造非 hash 和 zset类型
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.SADD);
        }

        /*
                取WaterSensor的什么作为key

                保留不同传感器的所有的信息
         */
        @Override
        public String getKeyFromData(WaterSensor data) {
            return "set_"+data.getId();
        }

        //取WaterSensor的什么作为value
        @Override
        public String getValueFromData(WaterSensor data) {
            return data.toString();
        }
    }

    private static class ListRedisMapper implements org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper<WaterSensor>
    {

        /*
                写命令

                RedisCommandDescription(RedisCommand redisCommand, String additionalKey)： 用于构造 hash 和 zset类型
                RedisCommandDescription(RedisCommand redisCommand)： 用于构造非 hash 和 zset类型
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.LPUSH);
        }

        /*
                取WaterSensor的什么作为key

                保留不同传感器的所有的信息
         */
        @Override
        public String getKeyFromData(WaterSensor data) {
            return "ws_"+data.getId();
        }

        //取WaterSensor的什么作为value
        @Override
        public String getValueFromData(WaterSensor data) {
            return data.toString();
        }
    }


    private static class StringRedisMapper implements org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper<WaterSensor>
    {

        /*
                写命令

                RedisCommandDescription(RedisCommand redisCommand, String additionalKey)： 用于构造 hash 和 zset类型
                RedisCommandDescription(RedisCommand redisCommand)： 用于构造非 hash 和 zset类型
         */
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.SET);
        }

        /*
                取WaterSensor的什么作为key

                只保留不同传感器的最新的信息
         */
        @Override
        public String getKeyFromData(WaterSensor data) {
            return data.getId();
        }

        //取WaterSensor的什么作为value
        @Override
        public String getValueFromData(WaterSensor data) {
            return data.toString();
        }
    }

}
